package org.ala.tree.lsm;

import java.io.IOException;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

import org.ala.tree.AbstractTree;
import org.ala.tree.lsm.memtable.SkipListMemTable;
import org.ala.tree.lsm.sstable.SSTableSearcher;
import org.ala.tree.lsm.sstable.SSTableWriter;
import org.ala.tree.lsm.tombstone.Tombstone;
import org.ala.tree.lsm.wal.Wal;

/**
 *
 * @author ala
 * @date 2025年5月25日
 */
public class LSMTree<V> extends AbstractTree<String, V> {
	/**
	 * 主配置
	 */
	protected LSMConfig config;
	/**
	 *  当前MemTable
	 */
	protected IMemTable<V> crtMemTable;
	/**
	 * 	只读的MemTable，和正在落盘的MemTable
	 */
	protected List<IMemTable<V>> readOnlyMemTables, saveMemTables;
	/**
	 *  墓碑
	 */
	protected Tombstone<String> tombstone;
	/**
	 * 	写入工具
	 */
	protected SSTableWriter<V> writer;
	/**
	 * 	搜索工具
	 */
	protected SSTableSearcher<V> searcher;
	
	
	public LSMTree(LSMConfig config, Comparator<String> comparator, Class<V> vclass) {
		super(comparator);
		this.vclass = vclass;
		
		this.config = config;
		crtMemTable = new SkipListMemTable<>(config, UUID.randomUUID().toString(), true);
		readOnlyMemTables = new LinkedList<>();
		tombstone = new Tombstone<>();
		writer = new SSTableWriter<>();
		searcher = new SSTableSearcher<>(config);
	}
	protected Class<V> vclass;

	
	@Override
	public boolean add(String k, V v) {
		try {
			_add(k, v);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
	}
	@SuppressWarnings("unchecked")
	protected void _add(String k, V v) {
		//	如果追加成功，检测当前内存块长度是否够固化
		if (crtMemTable.put(k, v)) {
			if (crtMemTable.size() >= config.memTableCapacity) {
				synchronized (LOCK_SAVE_DISK) {
					if (crtMemTable.size() >= config.memTableCapacity) {
						crtMemTable.setReadOnly();
						readOnlyMemTables.add(crtMemTable);
						crtMemTable = new SkipListMemTable<>(config, UUID.randomUUID().toString(), true);
						
						//	当只读的MemTable超过阈值时，执行落盘
						if (readOnlyMemTables.size() >= config.thresholdL0) {
//							saveDisk((Class<V>) v.getClass());
							saveMemTables = readOnlyMemTables;
							readOnlyMemTables = new LinkedList<>();
							_save(saveMemTables, vclass);
							saveMemTables = null;
						}
					}
				}
			}
		}
	}
	protected Object LOCK_SAVE_DISK = new Object();
	/**
	 * 	执行一次落盘
	 */
	public void saveDisk(Class<V> vclass) {
		synchronized (LOCK_SAVE_DISK) {
			crtMemTable.setReadOnly();
			readOnlyMemTables.add(crtMemTable);
			crtMemTable = new SkipListMemTable<>(config, UUID.randomUUID().toString(), true);
			
			saveMemTables = readOnlyMemTables;
			readOnlyMemTables = new LinkedList<>();
			_save(saveMemTables, vclass);
			saveMemTables = null;
		}
	}
	/**
	 * 只读的MemTable落盘
	 */
	protected void _save(List<IMemTable<V>> memTables, Class<V> vclass) {
		writer.save(config, memTables, vclass);
	}

	@Override
	public boolean remove(String k) {
		try {
			_remove(k);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		}
	}
	protected void _remove(String k) {
		//	如果k在crtMemTable中，直接删
		if (crtMemTable.remove(k) == null) {
			//	如果k在readOnlyMemTables中，直接删
			boolean inReadOnlyList = false;
			for (IMemTable<V> mt : readOnlyMemTables) {
				if (mt.remove(k) != null) {
					inReadOnlyList = true;
					break;
				}
			}
			//	如果也不在readOnly中，记录墓碑
			if (!inReadOnlyList) {
				tombstone.append(k);
			}
		}
	}
	

	@Override
	public boolean update(String k, V v) {
		return add(k,v);
	}

	@Override
	public V search(String k) {
		//	先从MemTable里找
		V v = crtMemTable.get(k);
		if (v != null) {return v;}
		//	再从只读MemTable里找
		for (IMemTable<V> mt : readOnlyMemTables) {
			v = mt.get(k);
			if (v != null) {return v;}
		}
		//	最后从sstable中逐级往上
		try {
			return searcher.search(k, 0, vclass);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	
	/**
	 * 	从wal文件还原只读MemTable
	 * @throws IOException 
	 */
	@SuppressWarnings({ "unchecked"})
	public void readOnlyMemTableFromWal(Class<V> vclass) throws IOException {
		IMemTable<V>[] mt = new IMemTable[1];
		Wal.iterator(config, vclass, (id, i, op, k, v) -> {
			if (mt[0] == null) {
				mt[0] = new SkipListMemTable<V>(config, id, false);
			}
			if (!mt[0].id().equals(id)) {
				mt[0].setReadOnly();
				readOnlyMemTables.add(mt[0]);
				mt[0] = new SkipListMemTable<V>(config, id, false);
			}
			
			switch (op) {
				case ADD: 
				case UPD: {mt[0].put(k, v); break;}
				case DEL: {mt[0].remove(k); break;}
			}
		});
	}

}
